package com.book.kafka.client.helper;

import com.book.kafka.client.record.JsonSerializer;
import com.book.kafka.client.record.KafkaRecord;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.util.HashSet;
import java.util.List;
import java.util.Set;

@Slf4j
@Service
public class KafkaHelper {

    @Value("${kafka.retry.topic:alikafka-topic_consume_failed_retry}")
    private String retryTopic;

    @Autowired
    private KafkaTemplate<String,byte[]> kafkaTemplate;

    public <T> void forwardList2RetryTopic(List<KafkaRecord<T>> kafkaRecords, String groupId){
        kafkaRecords.forEach(kafkaRecord -> this.forward2RetryTopic(kafkaRecord,groupId));
    }

    public <T> void forwardList2RetryTopic(List<KafkaRecord<T>> kafkaRecords){
        kafkaRecords.forEach(this::forward2RetryTopic);
    }

    public <T> void forward2RetryTopic(KafkaRecord<T> kafkaRecord){
        this.forward2RetryTopic(JsonSerializer.serialize(kafkaRecord));
        log.info("kafka message retry : {}", String.join("|", kafkaRecord.getTopic(),kafkaRecord.getKey(), String.valueOf(kafkaRecord.getRetryCount())));
    }

    public <T> void forward2RetryTopic(KafkaRecord<T> kafkaRecord, String groupId){
        Set<String> consumerGroupIds = kafkaRecord.getConsumerGroupIds();
        if (consumerGroupIds == null || consumerGroupIds.isEmpty()){
            consumerGroupIds = new HashSet<>();
        }
        consumerGroupIds.add(groupId);
        kafkaRecord.setConsumerGroupIds(consumerGroupIds);
        this.forward2RetryTopic(JsonSerializer.serialize(kafkaRecord));
    }

    public void forward2RetryTopic(List<ConsumerRecord<String, byte[]>> consumerRecords){
        consumerRecords.forEach(this::forward2RetryTopic);
    }

    public void forward2RetryTopic(ConsumerRecord<String, byte[]> consumerRecord){
        forward2RetryTopic(consumerRecord.value());
    }

    public void forward2RetryTopic(byte[] content){
        kafkaTemplate.send(retryTopic, content);
        kafkaTemplate.flush();
    }

    public boolean isContainConsumerGroupId(KafkaRecord<?> kafkaRecord, String groupId){
        Set<String> consumerGroupIds = kafkaRecord.getConsumerGroupIds();
        return consumerGroupIds == null || consumerGroupIds.isEmpty() || consumerGroupIds.contains(groupId);
    }

    public <T> KafkaRecord<T> consumerRecord2KafkaRecord(ConsumerRecord<String, byte[]> consumerRecord, Class<T> clazz){
        return new KafkaRecord<>(
                consumerRecord.topic(),
                0,
                consumerRecord.key(),
                consumerRecord.timestamp(),
                JsonSerializer.deserialize(consumerRecord.value(), clazz),
                null);
    }

}
